#ifndef __M_BROKER_H__
#define __M_BROKER_H__

// 引入muduo库组件（网络、协议编解码、日志等）
#include "muduo/proto/codec.h"      // Protobuf编解码器
#include "muduo/proto/dispatcher.h" // Protobuf消息分发器
#include "muduo/base/Logging.h"     // 日志工具
#include "muduo/base/Mutex.h"       // 互斥锁
#include "muduo/net/EventLoop.h"    // 事件循环
#include "muduo/net/TcpServer.h"    // TCP服务器
// 引入项目内部组件
#include "../mqcommon/mq_threadpool.hpp" // 线程池
#include "../mqcommon/mq_msg.pb.h"       // 消息结构定义（Protobuf）
#include "../mqcommon/mq_proto.pb.h"     // 协议请求/响应定义（Protobuf）
#include "../mqcommon/mq_logger.hpp"     // 日志工具
#include "mq_connection.hpp"             // 连接管理
#include "mq_consumer.hpp"               // 消费者管理
#include "mq_host.hpp"                   // 虚拟主机

namespace zdfmq
{
// 宏定义：元数据数据库文件路径和虚拟主机名
#define DBFILE "/meta.db"        // 元数据存储文件（如交换机、队列配置）
#define HOSTNAME "MyVirtualHost" // 默认虚拟主机名

    /**
     * @brief 消息队列服务器核心类
     * 负责启动TCP服务、处理客户端连接、分发请求到对应组件
     */
    class Server
    {
    public:
        // 定义Protobuf消息智能指针类型
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;

        /**
         * @brief 构造函数：初始化服务器核心组件
         * @param port 监听端口
         * @param basedir 数据存储基础路径
         */
        Server(int port, const std::string &basedir) : // 初始化TCP服务器（绑定事件循环、监听地址、服务器名、端口复用）
                                                       _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port),
                                                               "Server", muduo::net::TcpServer::kReusePort),
                                                       // 初始化消息分发器（设置未知消息处理回调）
                                                       _dispatcher(std::bind(&Server::onUnknownMessage, this, std::placeholders::_1,
                                                                             std::placeholders::_2, std::placeholders::_3)),
                                                       // 初始化Protobuf编解码器（绑定分发器的消息处理函数）
                                                       _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                                                                                                        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
                                                       // 初始化虚拟主机（包含元数据存储）
                                                       _virtual_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),
                                                       // 初始化消费者管理器
                                                       _consumer_manager(std::make_shared<ConsumerManager>()),
                                                       // 初始化连接管理器
                                                       _connection_manager(std::make_shared<ConnectionManager>()),
                                                       // 初始化线程池
                                                       _threadpool(std::make_shared<threadpool>())
        {

            // 初始化历史队列的消费者管理结构（从虚拟主机加载所有队列）
            QueueMap qm = _virtual_host->allQueues();
            for (auto &q : qm)
            {
                _consumer_manager->initQueueConsumer(q.first); // 为每个队列初始化消费者管理
            }

            // 注册所有协议请求的处理函数到分发器
            // 信道操作
            _dispatcher.registerMessageCallback<zdfmq::openChannelRequest>(
                std::bind(&Server::onOpenChannel, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<zdfmq::closeChannelRequest>(
                std::bind(&Server::onCloseChannel, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));

            // 交换机操作
            _dispatcher.registerMessageCallback<zdfmq::declareExchangeRequest>(
                std::bind(&Server::onDeclareExchange, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<zdfmq::deleteExchangeRequest>(
                std::bind(&Server::onDeleteExchange, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));

            // 队列操作
            _dispatcher.registerMessageCallback<zdfmq::declareQueueRequest>(
                std::bind(&Server::onDeclareQueue, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<zdfmq::deleteQueueRequest>(
                std::bind(&Server::onDeleteQueue, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));

            // 队列与交换机绑定
            _dispatcher.registerMessageCallback<zdfmq::queueBindRequest>(
                std::bind(&Server::onQueueBind, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<zdfmq::queueUnBindRequest>(
                std::bind(&Server::onQueueUnBind, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));

            // 消息操作
            _dispatcher.registerMessageCallback<zdfmq::basicPublishRequest>(
                std::bind(&Server::onBasicPublish, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<zdfmq::basicAckRequest>(
                std::bind(&Server::onBasicAck, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));

            // 消息订阅
            _dispatcher.registerMessageCallback<zdfmq::basicConsumeRequest>(
                std::bind(&Server::onBasicConsume, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<zdfmq::basicCancelRequest>(
                std::bind(&Server::onBasicCancel, this, std::placeholders::_1,
                          std::placeholders::_2, std::placeholders::_3));

            // 设置TCP服务器的消息回调（收到数据后交给编解码器处理）
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 设置TCP连接回调（连接建立/断开时触发）
            _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
        }

        /**
         * @brief 启动服务器
         */
        void start()
        {
            _server.start();  // 启动TCP服务器
            _baseloop.loop(); // 进入事件循环（阻塞）
        }

    private:
        // -------------------------- 协议请求处理函数 --------------------------
        /**
         * @brief 处理打开信道请求
         */
        void onOpenChannel(const muduo::net::TcpConnectionPtr &conn,
                           const openChannelRequestPtr &message,
                           muduo::Timestamp)
        {
            // 获取当前TCP连接对应的Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            { // 连接不存在，关闭连接
                DLOG("打开信道时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            // 调用Connection的openChannel处理请求
            return mconn->openChannel(message);
        }

        /**
         * @brief 处理关闭信道请求
         */
        void onCloseChannel(const muduo::net::TcpConnectionPtr &conn,
                            const closeChannelRequestPtr &message,
                            muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("关闭信道时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            return mconn->closeChannel(message);
        }

        /**
         * @brief 处理声明交换机请求
         */
        void onDeclareExchange(const muduo::net::TcpConnectionPtr &conn,
                               const declareExchangeRequestPtr &message,
                               muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("声明交换机时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            // 获取请求对应的信道，调用信道的declareExchange处理
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("声明交换机时，没有找到信道！");
                return;
            }
            return cp->declareExchange(message);
        }

        /**
         * @brief 处理删除交换机请求
         */
        void onDeleteExchange(const muduo::net::TcpConnectionPtr &conn,
                              const deleteExchangeRequestPtr &message,
                              muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("删除交换机时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("删除交换机时，没有找到信道！");
                return;
            }
            return cp->deleteExchange(message);
        }

        /**
         * @brief 处理声明队列请求
         */
        void onDeclareQueue(const muduo::net::TcpConnectionPtr &conn,
                            const declareQueueRequestPtr &message,
                            muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("声明队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("声明队列时，没有找到信道！");
                return;
            }
            return cp->declareQueue(message);
        }

        /**
         * @brief 处理删除队列请求
         */
        void onDeleteQueue(const muduo::net::TcpConnectionPtr &conn,
                           const deleteQueueRequestPtr &message,
                           muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("删除队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("删除队列时，没有找到信道！");
                return;
            }
            return cp->deleteQueue(message);
        }

        /**
         * @brief 处理队列绑定请求
         */
        void onQueueBind(const muduo::net::TcpConnectionPtr &conn,
                         const queueBindRequestPtr &message,
                         muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("队列绑定时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列绑定时，没有找到信道！");
                return;
            }
            return cp->queueBind(message);
        }

        /**
         * @brief 处理队列解绑请求
         */
        void onQueueUnBind(const muduo::net::TcpConnectionPtr &conn,
                           const queueUnBindRequestPtr &message,
                           muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("队列解除绑定时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列解除绑定时，没有找到信道！");
                return;
            }
            return cp->queueUnBind(message);
        }

        /**
         * @brief 处理消息发布请求
         */
        void onBasicPublish(const muduo::net::TcpConnectionPtr &conn,
                            const basicPublishRequestPtr &message,
                            muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("发布消息时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("发布消息时，没有找到信道！");
                return;
            }
            return cp->basicPublish(message);
        }

        /**
         * @brief 处理消息确认请求（Ack）
         */
        void onBasicAck(const muduo::net::TcpConnectionPtr &conn,
                        const basicAckRequestPtr &message,
                        muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("确认消息时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("确认消息时，没有找到信道！");
                return;
            }
            return cp->basicAck(message);
        }

        /**
         * @brief 处理消息订阅请求
         */
        void onBasicConsume(const muduo::net::TcpConnectionPtr &conn,
                            const basicConsumeRequestPtr &message,
                            muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("队列消息订阅时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列消息订阅时，没有找到信道！");
                return;
            }
            return cp->basicConsume(message);
        }

        /**
         * @brief 处理取消订阅请求
         */
        void onBasicCancel(const muduo::net::TcpConnectionPtr &conn,
                           const basicCancelRequestPtr &message,
                           muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("队列消息取消订阅时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列消息取消订阅时，没有找到信道！");
                return;
            }
            return cp->basicCancel(message);
        }

        /**
         * @brief 处理未知消息（未注册的请求类型）
         */
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,
                              const MessagePtr &message,
                              muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown(); // 关闭连接
        }

        /**
         * @brief 处理TCP连接事件（建立/断开）
         */
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            { // 连接建立
                // 创建新的Connection对象并加入管理
                _connection_manager->newConnection(_virtual_host, _consumer_manager, _codec, conn, _threadpool);
            }
            else
            { // 连接断开
                // 从管理器中移除连接
                _connection_manager->delConnection(conn);
            }
        }

    private:
        // 核心组件
        muduo::net::EventLoop _baseloop;        // 主事件循环
        muduo::net::TcpServer _server;          // TCP服务器对象
        ProtobufDispatcher _dispatcher;         // 消息分发器（路由请求到对应处理函数）
        ProtobufCodecPtr _codec;                // Protobuf编解码器
        VirtualHost::ptr _virtual_host;         // 虚拟主机（管理交换机、队列等元数据）
        ConsumerManager::ptr _consumer_manager; // 消费者管理器
        ConnectionManager::ptr _connection_manager;
        threadpool::ptr _threadpool;
    };
}
#endif